package com.ibm.cps.implementations;

import java.util.HashMap;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.KryoSerializable;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.ibm.cps.redis.RuleStateRedis;
import com.ibm.util.exception.CPSException;

public class RuleState implements KryoSerializable {

	String ruleId;
	RuleStateRedis redis = null;
	HashMap<String, Boolean> timeSeriesState;
	
	public RuleState(){}

	public RuleState(String ruleId, RuleStateRedis redis) throws CPSException {
		this.ruleId = ruleId;
		this.redis = redis;
		timeSeriesState = redis.readRuleState(ruleId);
		if (timeSeriesState == null) {
			timeSeriesState = new HashMap<String, Boolean>();
		}
	}

	public String constructKey(String dataSourceId, long timeSeries) {
		return dataSourceId + "_" + timeSeries;
	}

	public boolean readRuleState(Comparable timeSeries) throws CPSException {
		String key = timeSeries + "";
		if (timeSeriesState.containsKey(key)) {
			return timeSeriesState.get(key);
		} else {
			timeSeriesState.put(key, false);
			redis.updateRuleState(ruleId, key, false);
			return false;
		}
	}

	public void setRuleState(Comparable timeSeries, boolean state)
			throws CPSException {
		String key = timeSeries + "";
		timeSeriesState.put(key, state);
		redis.updateRuleState(ruleId, key, state);
	}

	@Override
	public void write(Kryo kryo, Output output) {
		output.writeString(ruleId);
		kryo.writeObject(output, timeSeriesState);
		String host = redis.getRedisHost();
		int port = redis.getRedisPort();
		output.writeString(host);
		output.writeInt(port);
	}

	@Override
	public void read(Kryo kryo, Input input) {
		this.ruleId = input.readString();
		this.timeSeriesState = kryo.readObject(input, HashMap.class);
		String host = input.readString();
		int port = input.readInt();
		this.redis = RuleStateRedis.getRuleStateRedis(host, port);
	}

}
